package util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;

public class MyHBaseUtil {
    static Configuration con = HBaseConfiguration.create();
    static Configuration conf = MyProps.setConf(con);
    static Connection connection;
    static HBaseAdmin admin;
    static Table t;

    static {
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = (HBaseAdmin)connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //获取 conn
    public static Connection getConf(){
        //创建HBase的配置对象
        Configuration conf = HBaseConfiguration.create();
        //设置hbase配置属性
        conf.set("hbase.zookeeper.quorum","hadoop106,hadoop107,hadoop108");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        Connection connection=null;
        //通过连接函数，创建连接对象
        try {
            connection = ConnectionFactory.createConnection(conf);

        } catch (IOException e) {
            e.printStackTrace();
        }
        return connection;
    }

    //建表
    public static void build_Table(String tableName,List<String> FamilyNames) throws Exception {
       TableDescriptorBuilder buider = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
       for (String columnName : FamilyNames) {
           ColumnFamilyDescriptor info = ColumnFamilyDescriptorBuilder.of(Bytes.toBytes(columnName));
           buider.setColumnFamily(info);
       }
       TableDescriptor build = buider.build();
       admin.createTable(build);
        System.out.println("____build_done____");
   }

    //插入一条数据
    public static void insert_Row(String tableName,String row,String Family,String qualifier,String value) throws Exception {
        t = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(row));
        Put put1 = put.addColumn(Bytes.toBytes(Family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
        t.put(put1);
        System.out.println("____insert_Row_done____");
    }

    //插入num条数据
    public  static  void insert_Batch(String tableName,String row,String Family,String qualifier,String value,Integer num) throws Exception {
        t = connection.getTable(TableName.valueOf(tableName));
         List<Put> list=new ArrayList<>();
        for (int i = 0; i < num; i++) {
            String s = UUID.randomUUID().toString().replaceAll("-", "");
            Put puts = new Put(Bytes.toBytes(s));
            Put putss = puts.addColumn(Bytes.toBytes(Family), Bytes.toBytes(qualifier), Bytes.toBytes(value+i));
            list.add(putss);
        }
        t.put(list);
        System.out.println("____insert_Batch_done____");
    }

    //删除表
    public static void drop_Table(String tableName) throws Exception {
        if (admin.tableExists(TableName.valueOf(tableName))){
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
            System.out.println("____drop_Table_done____");
        }else {
            System.out.println("____no_such_Table_found____");
        }

    }

    //删除一条数据
    public static void  delete_Row(String tableName,String row) throws Exception {
        t = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(row));
        t.delete(delete);
        System.out.println("____delete_Row_done____");
    }

    //特定列过滤查询
    public static void scan_Filter(String tableName,String Family,String qualifier,CompareOperator compare,byte[] value) throws Exception {
        t = connection.getTable(TableName.valueOf(tableName));
        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(Family), Bytes.toBytes(qualifier), compare, value);
        Scan scan = new Scan();
        Scan scan1 = scan.setFilter(filter);
        ResultScanner scanner = t.getScanner(scan1);
        Iterator<Result> iterator = scanner.iterator();
        while (iterator.hasNext()){
            Cell[] cells = iterator.next().rawCells();
            System.out.println("____"+new String(CellUtil.cloneRow(cells[0]))+"____");
            for (Cell cell : cells) {
                System.out.print(new String(CellUtil.cloneRow(cell)));
                System.out.print(" - ");
                System.out.print(new String(CellUtil.cloneFamily(cell)));
                System.out.print(" - ");
                System.out.print(new String(CellUtil.cloneQualifier(cell)));
                System.out.print(" - ");
                System.out.println(new String(CellUtil.cloneValue(cell)));
            }

        }
        System.out.println("____scan_Filter_done____");
    }
    //查询一条数据
    public static void scan_Row(String tableName,String row)throws Exception{
        t = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(row));
        Result result = t.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.print(new String(CellUtil.cloneRow(cell)));
            System.out.print(" - ");
            System.out.print(new String(CellUtil.cloneFamily(cell)));
            System.out.print(" - ");
            System.out.print(new String(CellUtil.cloneQualifier(cell)));
            System.out.print(" - ");
            System.out.println(new String(CellUtil.cloneValue(cell)));
        }
        System.out.println("____scan_Row_done____");
    }
    //区间查询数据
    public static void scan_Rows(String tableName,String row1,String row2)throws Exception{
        t = connection.getTable(TableName.valueOf(tableName));
        Scan sc=new Scan(Bytes.toBytes(row1),Bytes.toBytes(row2));
        ResultScanner scanner = t.getScanner(sc);
        Iterator<Result> iterator = scanner.iterator();
        System.out.println("____前闭后开____");
        while (iterator.hasNext()){
            Result next = iterator.next();
            Cell[] cells = next.rawCells();
            System.out.println("____"+new String(CellUtil.cloneRow(cells[0]))+"____");
            for (Cell cell : cells) {
                System.out.print(new String(CellUtil.cloneRow(cell)));
                System.out.print(" - ");
                System.out.print(new String(CellUtil.cloneFamily(cell)));
                System.out.print(" - ");
                System.out.print(new String(CellUtil.cloneQualifier(cell)));
                System.out.print(" - ");
                System.out.println(new String(CellUtil.cloneValue(cell)));
            }
        }
        System.out.println("____scan_Rows_done____");
    }
    //查询一条特定列族数据
    public static void get_value_by_family(String tableName,String row,String family)throws Exception{
        t = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(row));
        get.addFamily(Bytes.toBytes(family));
        Result result = t.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.print(new String(CellUtil.cloneRow(cell)));
            System.out.print(" - ");
            System.out.print(new String(CellUtil.cloneFamily(cell)));
            System.out.print(" - ");
            System.out.print(new String(CellUtil.cloneQualifier(cell)));
            System.out.print(" - ");
            System.out.println(new String(CellUtil.cloneValue(cell)));
        }
        System.out.println("____get_value_by_family_done____");
    }

    //查询一条特定列数据
    public static void get_value_by_qualifier(String tableName,String row,String family,String qualifier)throws Exception{
        t = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(row));
        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
        Result result = t.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.print(new String(CellUtil.cloneRow(cell)));
            System.out.print(" - ");
            System.out.print(new String(CellUtil.cloneFamily(cell)));
            System.out.print(" - ");
            System.out.print(new String(CellUtil.cloneQualifier(cell)));
            System.out.print(" - ");
            System.out.println(new String(CellUtil.cloneValue(cell)));
        }
        System.out.println("____get_value_by_qualifier_done____");
    }

    //全查某表
    public static void  scan_All(String tableName) throws Exception {
        t = connection.getTable(TableName.valueOf(tableName));
        Scan sc=new Scan();
        ResultScanner scanner = t.getScanner(sc);
        Iterator<Result> iterator = scanner.iterator();
        while (iterator.hasNext()){
            Result next = iterator.next();
            Cell[] cells = next.rawCells();
            System.out.println("____"+new String(CellUtil.cloneRow(cells[0]))+"____");
            for (Cell cell : cells) {
                System.out.print(new String(CellUtil.cloneRow(cell)));
                System.out.print(" - ");
                System.out.print(new String(CellUtil.cloneFamily(cell)));
                System.out.print(" - ");
                System.out.print(new String(CellUtil.cloneQualifier(cell)));
                System.out.print(" - ");
                System.out.println(new String(CellUtil.cloneValue(cell)));
            }
        }
        System.out.println("____scan_All_done____");
    }
    //查看所有表
    public static  void list() throws Exception {
        TableName[] tableNames = admin.listTableNames();
        for (TableName tableName : tableNames) {
            System.out.println(tableName.toString());
        }

    }
    //查看所有表结构
    public static  void desc_Table(String tableName) throws Exception {
        List<TableDescriptor> tableDescriptors = admin.listTableDescriptors();
        Iterator<TableDescriptor> iterator = tableDescriptors.iterator();
        while (iterator.hasNext()){
            TableDescriptor next = iterator.next();
            if (next.getTableName().toString().equals(tableName)){
                System.out.println(next);
            }
        }
        System.out.println("____list_done____");

    }


    //关流
    public static  void stop() throws Exception {
        connection.close();
        System.out.println("____connection_close_done____");
    }
}
